package com.rrd.dw.mr.gzip.kafka;

import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.zitacr.Dishonesty4Data;
import com.ucredit.babel.common.model.zitacr.ZitacrDishonesty4Record;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 泽泰-失信被执行人4.0
 *
 * @author renzhanzheng
 */

public class ZitacrDishonesty4Parse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 泽泰-失信被执行人4.0
     *
     * @param model
     * @throws IOException
     * @throws InterruptedException
     * @return
     */
    @Override
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {
        ZitacrDishonesty4Record obj = (ZitacrDishonesty4Record) model;
        List<Dishonesty4Data> data = obj.getData() == null ? new ArrayList<Dishonesty4Data>() : obj.getData();

        StringBuilder sb = getBaseInfo(obj, successful);

        //返回信息
        sb.append(obj.getMsg() == null ? "" : obj.getMsg()).append(CRT001);
        //返回订单号
        sb.append(obj.getOrderNo() == null ? "" : obj.getOrderNo()).append(CRT001);
        //返回状态码
        sb.append(obj.getCode() == null ? "" : obj.getCode()).append(CRT001);

        StringBuilder detailSb = new StringBuilder(500);
        int i = 0;
        for (Dishonesty4Data d : data) {
            i++;
            detailSb.delete(0, sb.length());
            //序号
            detailSb.append(i).append(CRT001);
            //案号
            detailSb.append(d.getCaseCode() == null ? "" : d.getCaseCode()).append(CRT001);
            //未履行
            detailSb.append(d.getUnPerformPart() == null ? "" : d.getUnPerformPart()).append(CRT001);
            //性别
            detailSb.append(d.getSex() == null ? "" : d.getSex()).append(CRT001);
            //年龄
            detailSb.append(d.getAge() == null ? "" : d.getAge()).append(CRT001);
            //法定代表人或者负责人姓名
            detailSb.append(d.getBussinessEntity() == null ? "" : d.getBussinessEntity()).append(CRT001);
            //做出执行依据单位
            detailSb.append(d.getGistUnit() == null ? "" : d.getGistUnit()).append(CRT001);
            //发布日期
            detailSb.append(d.getPublishDate() == null ? "" : d.getPublishDate()).append(CRT001);
            //立案时间
            detailSb.append(d.getRegDate() == null ? "" : d.getRegDate()).append(CRT001);
            //主体代码
            detailSb.append(d.getEntityId() == null ? "" : d.getEntityId()).append(CRT001);
            //类型
            detailSb.append(d.getPartyTypeName() == null ? "" : d.getPartyTypeName()).append(CRT001);
            //命中类型(失信被执行人、关联案件、曾用名)
            detailSb.append(d.getRelateType() == null ? "" : d.getRelateType()).append(CRT001);
            //已履行
            detailSb.append(d.getPerformedPart() == null ? "" : d.getPerformedPart()).append(CRT001);
            //失信类型
            detailSb.append(d.getDisruptTypeName() == null ? "" : d.getDisruptTypeName()).append(CRT001);
            //被执行人的履行情况
            detailSb.append(d.getPerformance() == null ? "" : d.getPerformance()).append(CRT001);
            //法院
            detailSb.append(d.getCourtName() == null ? "" : d.getCourtName()).append(CRT001);
            //省份
            detailSb.append(d.getAreaName() == null ? "" : d.getAreaName()).append(CRT001);
            //主体名称
            detailSb.append(d.getEntityName() == null ? "" : d.getEntityName()).append(CRT001);
            //生效法律文书确定的义务
            detailSb.append(d.getDuty() == null ? "" : d.getDuty()).append(CRT001);
            //执行依据文号
            detailSb.append(d.getGistId() == null ? "" : d.getGistId()).append(CRT001);
            //下架状态：已下架；未下架
            detailSb.append(d.getStatus() == null ? "" : d.getStatus()).append(CRT001);

            key.set(getKey());
            val.set((sb.toString() + detailSb.toString()).replace("\n", "").replace("\r", " "));
            context.write(key, val);
        }

        if (i == 0) {
            key.set(getKey());
            val.set(sb.toString().replace("\n", " ").replace("\r", " "));
            context.write(key, val);
        }

        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_zetai_dishonesty_inc";
    }

    @Override
    public String getType() {
        return RecordType.ZITACR_DISHONESTY4_RECORD.toString();
    }

    @Override
    public String getComment() {
        return "泽泰-失信被执行人4.0";
    }
}
